
package com.lee.lossycounting; 
 
import com.alibaba.fastjson.JSONArray; 
import com.google.common.collect.Lists; 
import com.lee.enumeration.FinishQualifierEnum; 
import com.lee.info.CkFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.utils.*; 
import org.apache.flink.api.common.functions.FlatMapFunction; 
import org.apache.flink.util.Collector; 
 
import java.util.ArrayList; 
import java.util.HashMap; 
import java.util.List; 
import java.util.Map; 
 
public class DCountHandleFunction implements FlatMapFunction<List<String>, CkFlowInfo> { 
 
 
    @Override
    public void flatMap(List<String> bdsCkFlowList, Collector<CkFlowInfo> collector)  {
 
        try {
 
            DMapUtils.addIndex();
            DCountUtils.addIndex();
            Logger.debug(DMapUtils.getIndex(),"/",DCountUtils.getIndex(),"LossyCounting DCountHandleFunction -> index + 1, bdsCkFlowList ", bdsCkFlowList);
 
 
            //判断是否到了块结尾
            long blockSize = ConstantsUtils.DATA_BLOCK_SIZE;
            if (DMapUtils.getIndex() > 0 && DMapUtils.getIndex() % blockSize == 0) {
                //等待DMap表不再更新
                long waitTimes = 60;
                long finishIndex = 0;
                long lastFinishIndex = -1;
                while(waitTimes > 0 && finishIndex != lastFinishIndex){
                    waitTimes -= 1;
                    lastFinishIndex = finishIndex;
                    finishIndex = 0;
                    for(int i=0;i<FinishQualifierEnum.values().length;i++){
                        finishIndex += DMapUtils.getFinishIndex(FinishQualifierEnum.values()[i].getCode());
                    }
 
                    Logger.info("LossyCounting DCountHandleFunction ->  表处理未完成,进度 "+finishIndex+"，剩余",waitTimes,"次尝试，等待一段时间后重新获取 ");
                    Thread.sleep(10000);
                }
                //获取SupportData所有的key
                List<String> allKey = SupportDataUtils.getAllKey();
                Logger.info(DMapUtils.getIndex(),"/",DCountUtils.getIndex(),"LossyCounting DCountHandleFunction -> 查询得到supportData表一共有key的数量", allKey.size());
                //将所有key的支持度计数-1，删除计数为0的值
                for (String key : allKey) {
 
                    SupportDataUtils.delSupportNum(key);
                }
                ;
 
            }
            CkFlowInfo ckFlowInfo = new CkFlowInfo();
            long index = DMapUtils.getIndex();
            long cindex = DCountUtils.getIndex();
            ckFlowInfo.setDIndex(index);
            ckFlowInfo.setDCIndex(cindex);
            ckFlowInfo.setCkList(bdsCkFlowList);
            collector.collect(ckFlowInfo);
            Logger.info(DMapUtils.getIndex(),"/",DCountUtils.getIndex(),"END LossyCounting DCountHandleFunction -> index + 1, bdsCkFlowList ", bdsCkFlowList);
 
        }catch (Exception e){
            Logger.error("DCountHandleFunction -> 初始化数据报错：",e);
 
        }
    }
 
 
 
} 
